package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @packageName:com.twoandone.tms.realtime.app.dwd
 * @ClassName:DwdTransportTaskProcess
 * @Description:
 * @author:
 * @date 2023/3/10 18:35
 **/
public class DwdTransportTaskProcess {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(10));
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_transport_task_process"));
//      运输任务表
        Table transportTask = tableEnv.sqlQuery("select \n" +
                "\t`data`['id'] id,\n" +
                "\t`data`['shift_id'] shift_id,\n" +
                "\t`data`['line_id'] line_id,\n" +
                "\t`data`['start_org_id'] start_org_id,\n" +
                "\t`data`['start_org_name'] start_org_name,\n" +
                "\t`data`['end_org_id'] end_org_id,\n" +
                "\t`data`['end_org_name'] end_org_name,\n" +
                "\t`data`['status'] status,\n" +
                "\t`data`['order_num'] order_num,\n" +
                "\t`data`['driver1_emp_id'] driver1_emp_id,\n" +
                "\t`data`['driver1_name'] driver1_name,\n" +
                "\t`data`['driver2_emp_id'] driver2_emp_id,\n" +
                "\t`data`['driver2_name'] driver2_name, \n" +
                "\t`data`['truck_id'] truck_id,\n" +
                "\t`data`['truck_no'] truck_no, \n" +
                "\t`data`['actual_start_time'] actual_start_time,\n" +
                "\t`data`['actual_end_time'] actual_end_time,\n" +
                "\t`data`['actual_distance'] actual_distance,\n" +
                "\t`data`['create_time'] create_time,\n" +
                "\tts,\n" +
                "\tproc_time\n" +
                "from `topic_db`\n" +
                "where `table` = 'transport_task'\n" +
                "\t and `type` = 'insert'");
        tableEnv.createTemporaryView("transport_task", transportTask);

        Table transportTaskProcess = tableEnv.sqlQuery("select \n" +
                "\t`data`['id'] transport_task_process_id,\n" +
                "\t`data`['transport_task_id'] transport_task_id,\n" +
                "\t`data`['cur_distance'] cur_distance,\n" +
                "\t`data`['line_distance'] line_distance " +
                " from topic_db\n" +
                " where `table` = 'transport_task_process' \n" +
                " \t  and `type` = 'insert'");

        tableEnv.createTemporaryView("transport_task_process", transportTaskProcess);

        tableEnv.executeSql(MySqlUtil.getBaseDicLookUpDDL());

        Table joinTable = tableEnv.sqlQuery("select \n" +
                " transport_task.id,\n" +
                " transport_task.shift_id,\n" +
                " transport_task.line_id,\n" +
                " transport_task.start_org_id,\n" +
                " transport_task.start_org_name,\n" +
                " transport_task.end_org_id,\n" +
                " transport_task.end_org_name,\n" +
                " transport_task.status,\n" +
                " dic.name,\n" +
                " cast(dic.parent_id as string),\n" +
                " transport_task.order_num,\n" +
                " transport_task.driver1_emp_id,\n" +
                " transport_task.driver1_name,\n" +
                " transport_task.driver2_emp_id,\n" +
                " transport_task.driver2_name,\n" +
                " transport_task.truck_id,\n" +
                " transport_task.truck_no,\n" +
                " transport_task.actual_start_time,\n" +
                " transport_task.actual_end_time,\n" +
                " transport_task.actual_distance,\n" +
                " transport_task.create_time,\n" +
                " transport_task.ts,\n" +
                " transport_task_process.cur_distance,\n" +
                " transport_task_process.line_distance\n" +
                "from transport_task \n" +
                "join transport_task_process\n" +
                "on transport_task.id = transport_task_process.transport_task_id\n" +
                "join base_dic for SYSTEM_TIME AS OF transport_task.proc_time AS dic\n" +
                " ON cast(transport_task.status as bigint) = dic.id ");
        tableEnv.createTemporaryView("join_table", joinTable);


        tableEnv.executeSql(" CREATE TABLE dwd_transport_task_process (\n" +
                "\tid string,\n" +
                "\tshift_id string,\n" +
                "\tline_id string,\n" +
                "\tstart_org_id string,\n" +
                "\tstart_org_name string,\n" +
                "\tend_org_id string,\n" +
                "\tend_org_name string,\n" +
                "\tstatus string,\n" +
                "\tname string,\n" +
                "\tparent_id string,\n" +
                "\torder_num string,\n" +
                "\tdriver1_emp_id string,\n" +
                "\tdriver1_name string,\n" +
                "\tdriver2_emp_id string,\n" +
                "\tdriver2_name string,\n" +
                "\ttruck_id string,\n" +
                "\ttruck_no string,\n" +
                "\tactual_start_time string,\n" +
                "\tactual_end_time string,\n" +
                "\tactual_distance string,\n" +
                "\tcreate_time string,\n" +
                "\tts string,\n" +
                "\tcur_distance string,\n" +
                "\tline_distance string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_transport_task_process"));


        tableEnv.executeSql("insert into dwd_transport_task_process select * from join_table");

        tableEnv.executeSql("select * from join_table").print();
    }
}
